package cn.admin.configcenter.configservice.service;

import cn.admin.configcenter.configservice.util.ConfigKey;
import cn.admin.configcenter.configservice.util.RequestResponseUtil;
import cn.admin.configcenter.configservice.util.ServerConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.*;
import java.util.concurrent.*;

@Service
public class LongPullingService {

    private static final Logger LOG = LoggerFactory.getLogger(LongPullingService.class);
    public static final String LONG_PULLING_HEADER = "longPullingTimeout";
    public static final String LONG_PULLING_NO_HANG_UP_HEADER = "longPullingNoHangUp";
    private Map<String, Long> retainIps = new ConcurrentHashMap<>();

    @Autowired
    private ConfigService configService;

    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, r -> {
        Thread t = new Thread(r);
        t.setDaemon(true);
        t.setName(LongPullingService.class.getName());
        return t;
    });

    private final Queue<ClientLongPulling> allSubs = new ConcurrentLinkedQueue<>();

    public LongPullingService() {
        this.scheduler.scheduleWithFixedDelay(new StatTask(),0,10, TimeUnit.SECONDS);
    }

    public void onConfigChangeEvent(String configKey) {
        this.scheduler.execute(new DataChangeTask(configKey));
    }

    public Map<String, Long> getRetainIps() {
        return retainIps;
    }

    public String compareMd5ResultString(List<String> changedGroupKeys) throws UnsupportedEncodingException {
        if (changedGroupKeys == null) {
            return ServerConstants.NULL;
        } else {
            StringBuilder sb = new StringBuilder();
            for (String groupKey : changedGroupKeys) {
                String[] dataIdGroupId = ConfigKey.parseKey(groupKey);
                sb.append(dataIdGroupId[0]);
                sb.append(ServerConstants.WORD_SEPARATOR);
                sb.append(dataIdGroupId[1]);
                sb.append(ServerConstants.LINE_SEPARATOR);
            }
            return URLEncoder.encode(sb.toString(),"UTF-8");
        }
    }

    public List<String> getSubscribedConfigKeyByClientIp(String queryClientIp) {
        List<String> result = new ArrayList<>();
        Iterator<ClientLongPulling> it = this.allSubs.iterator();
        while (true) {
            ClientLongPulling clientLongPulling;
            String longPullingClientIp;
            do {
                if (!it.hasNext()) {
                    return result;
                }

                clientLongPulling = it.next();
                HttpServletRequest request = (HttpServletRequest) clientLongPulling.asyncContext.getRequest();
                longPullingClientIp = RequestResponseUtil.getRemoteIp(request);
            } while (!queryClientIp.equals(longPullingClientIp) && !longPullingClientIp.contains(queryClientIp));

            result.addAll(clientLongPulling.clientMd5Map.keySet());
        }
    }

    class ClientLongPulling implements Runnable {

        final AsyncContext asyncContext;
        final Map<String,String> clientMd5Map;
        final long createTime;
        final String ip;
        final String appName;
        final int probeRequestSize;
        final long timeoutTime;
        Future<?> asyncTimeoutFuture;

        @Override
        public void run() {
            this.asyncTimeoutFuture = LongPullingService.this.scheduler.schedule(() -> {
                try {
                    LongPullingService.this.getRetainIps().put(ClientLongPulling.this.ip,
                            System.currentTimeMillis());
                    LongPullingService.this.allSubs.remove(ClientLongPulling.this);
                    LOG.info("{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - ClientLongPulling.this.createTime,
                            "timeout",
                            RequestResponseUtil.getRemoteIp((HttpServletRequest)ClientLongPulling.
                                    this.asyncContext.getRequest()), "polling",
                            ClientLongPulling.this.clientMd5Map.size(),
                            ClientLongPulling.this.probeRequestSize);
                    ClientLongPulling.this.sendResponse(null);
                } catch (Exception e) {
                    LOG.error("long pulling error:" + e.getMessage(), e.getCause());
                }
            },this.timeoutTime,TimeUnit.MILLISECONDS);
            LongPullingService.this.allSubs.add(this);
        }

        void sendResponse(List<String> changedGroups) {
            if (this.asyncTimeoutFuture != null) {
                this.asyncTimeoutFuture.cancel(false);
            }
            this.generateResponse(changedGroups);
        }

        void generateResponse(List<String> changedGroups) {
            if (changedGroups == null) {
                this.asyncContext.complete();
            } else {
                HttpServletResponse response = (HttpServletResponse) this.asyncContext.getResponse();

                try {
                    String respString = LongPullingService.this.compareMd5ResultString(changedGroups);
                    RequestResponseUtil.setCommonResponseHeaders(response);
                    if (respString != null && !respString.equals(ServerConstants.NULL)) {
                        response.getWriter().println(respString);
                    } else {
                        response.getWriter().println("OK");
                    }

                    this.asyncContext.complete();
                } catch (IOException e) {
                    LOG.error(e.getMessage(),e);
                    this.asyncContext.complete();
                }
            }
        }

        ClientLongPulling(AsyncContext ac, Map<String, String> clientMd5Map,
                          String ip, int probeRequestSize, long timeoutTime, String appName) {
            this.asyncContext = ac;
            this.clientMd5Map = clientMd5Map;
            this.probeRequestSize = probeRequestSize;
            this.createTime = System.currentTimeMillis();
            this.ip = ip;
            this.timeoutTime = timeoutTime;
            this.appName = appName;
        }
    }

    class DataChangeTask implements Runnable {

        final String groupKey;
        final long changeTime = System.currentTimeMillis();

        DataChangeTask(String groupKey) {
            this.groupKey = groupKey;
        }

        @Override
        public void run() {
            try {
                Iterator<ClientLongPulling> it = LongPullingService.this.allSubs.iterator();

                while (it.hasNext()) {
                    ClientLongPulling clientSub = it.next();
                    if (clientSub.clientMd5Map.containsKey(this.groupKey)) {
                        LongPullingService.this.getRetainIps().put(clientSub.ip,
                                System.currentTimeMillis());
                        it.remove();
                        LongPullingService.LOG.info("{}|{}|{}|{}|{}|{}|{}",
                                System.currentTimeMillis() - this.changeTime,
                                "in-advance",
                                RequestResponseUtil.getRemoteIp((HttpServletRequest)clientSub.asyncContext.getRequest()),
                                "polling",
                                clientSub.clientMd5Map.size(),
                                clientSub.probeRequestSize,
                                this.groupKey);
                        clientSub.sendResponse(Collections.singletonList(this.groupKey));
                    }
                }
            } catch (Exception e) {
                LOG.error("data change error:" + e.getMessage(), e.getCause());
            }
        }
    }

    class StatTask implements Runnable {

        @Override
        public void run() {
            LOG.debug("[long-pulling] client count " + LongPullingService.this.allSubs.size());
        }
    }
}
